package com.example.demo.mq.producer;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageProducer;
import org.apache.activemq.AsyncCallback;

import javax.jms.*;
import java.util.UUID;

public class ActivemqAsyncProducer {
    static String BROCKER_URL = "tcp://127.0.0.1:61616";
    static String QUEUE_NAME = "queue_async";


    /**
     * 一、如何使用
     * 1、传统模式
     * 2、SpringBoot模式
     *
     * 二、保证消息的可靠性
     * 消息不丢失（生产者--1回调、broker--2持久化机制、消费者--3应答手动提交、）
     * 不被重复消费（消费者提交事务、幂等性）
     * 1.异步回调保证消息发出去、ack
     * 2.消息持久化（mq宕机消息不丢失）
     * 3.非事务手动签收需要手动应答、重试机制
     *
     * 持久化（Persistence）：通过将消息存储到磁盘上，即使在ActiveMQ服务宕机的情况下，消息也不会丢失。在重启服务后，这些消息可以重新入队并被消费者处理。
     * 事务（Transactions）：使用事务可以确保消息的发送和接收都在一个原子操作中完成。如果事务中的任何步骤失败，整个事务都会被回滚，从而保证消息不会丢失。
     * 消息确认（Acknowledgements）：消费者在成功处理消息后，会向ActiveMQ发送确认信息。这可以确保消息在被确认之前不会被删除，从而保证消息的传递是可靠的。
     * 消息重试（Message Redelivery）：在某些情况下，如果消费者无法成功处理消息，ActiveMQ可以将消息重新发送给消费者，直到处理成功或达到重试次数上限。
     *
     * 三、高级用法
     * 1、协议 【nio协议同步非阻塞】可以提高性能
     * 2、延时定时投递 【指定用同步发送，在未使用事务前提下发送持久化消息】
     * 3、死信队列【消费者重试一直重复消费没提交事务，那到达一个重试阈值，消息变成有毒消息，那么有毒消息就会进入死信队列】
     * 4、集群 提高mq可用性，但需要保证稳定性，可靠性
     */

    public static void main(String[] args) throws JMSException {
        // 1. 创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(BROCKER_URL);
        // 设置异步投递方式
        activeMQConnectionFactory.setUseAsyncSend(true);
        // 2. 获取连接 并启动
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        // 3. 创建session ,参数一：事务，参数二：签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4. 创建目的地,队列
        Queue queue = session.createQueue(QUEUE_NAME);
        // 5.创建消息生产者 类型为ActiveMQMessageProducer
        ActiveMQMessageProducer messageProducer = (ActiveMQMessageProducer) session.createProducer(queue);
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);  //消息持久化
        // 6.通过 messageProducer 发送3条消息
        for (int i = 1; i <= 3; i++) {
            TextMessage textMessage = session.createTextMessage("消息 ->" + i);
            textMessage.setJMSMessageID(UUID.randomUUID() + "--" + i);
            String messageID = textMessage.getJMSMessageID();
            // 发送消息，异步回调
            messageProducer.send(textMessage, new AsyncCallback() {
                public void onSuccess() {
                    System.out.println("消息确认发送成功。。。" + messageID);
                }

                public void onException(JMSException e) {
                    System.out.println("消息发送失败。。。" + messageID);
                }
            });

        }

        // 7.关闭资源
        messageProducer.close();
        session.close();
        connection.close();
    }
}
